package com.zy.demo;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.jupiter.api.Test;

import java.util.List;

public class SimpleMq {

    //NameServer地址 ,但是broker地址的端口也需要开放
    private static final String NAMESERVER_ADDR = "192.168.200.120:9876";

    /**
     * 生产者测试
     */
    @Test
    public void simpleProduce() throws Exception {

        //创建生产者,指定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("test_producer_group");
        //指定NameServer地址
        producer.setNamesrvAddr(NAMESERVER_ADDR);
        //启动生产者
        producer.start();
        //创建消息
        Message message = new Message("testTopic", "Hello,RocketMQ!".getBytes());
        //发送同步消息
        SendResult result = producer.send(message);
        //返回结果
        System.out.println(result);
        //关闭生产者
        producer.shutdown();
    }

    /**
     * 消费者测试
     */
    @Test
    public void simpleConsume() throws Exception {
        //创建消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_consumer_group");
        //指定nameServer地址
        consumer.setNamesrvAddr(NAMESERVER_ADDR);
        //订阅主题
        consumer.subscribe("testTopic", "*");
        //注册监听
        consumer.registerMessageListener((List<MessageExt> msgList, ConsumeConcurrentlyContext context)->{
            System.out.println(new String(msgList.get(0).getBody()));
            System.out.println(context);
            //这个返回状态是消费成功还是失败，如果失败，则消息会重新发送
            //这里返回成功，表示消费成功
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        //启动消费者，这里和生产者不一样，先要注册监听后启动
        consumer.start();

        System.in.read();
    }

}
